package endpoint

import (
	"github.com/darwinOrg/go-common/constants"
	redisdk "github.com/darwinOrg/go-redis"
	"github.com/go-mysql-org/go-mysql/canal"
	"github.com/go-mysql-org/go-mysql/mysql"
	"github.com/juju/errors"
	"myc/global"
	"myc/metrics"
	"myc/model"
	"myc/util/alarm"
	"myc/util/logs"
	"myc/util/stringutil"
	"time"
)

var redisStreamMessageExpiration = 48 * time.Hour

type RedisStreamEndpoint struct {
	*RedisEndpoint
}

func newRedisStreamEndpoint() *RedisStreamEndpoint {
	r := &RedisStreamEndpoint{}
	r.RedisEndpoint = NewRedisEndpoint()
	return r
}

func (s *RedisStreamEndpoint) Name() string {
	return "redis_stream"
}

func (s *RedisStreamEndpoint) Consume(from mysql.Position, rows []*model.RowRequest) error {
	for _, row := range rows {
		rule, _ := global.RuleIns(row.RuleKey)
		if rule.TableColumnSize != len(row.Row) {
			logs.Warnf("%s schema mismatching", row.RuleKey)
			continue
		}

		metrics.UpdateActionNum(row.Action, row.RuleKey)

		err := s.doRuleConsume(row, rule)
		if err != nil {
			alarm.EmergencyAlarm(err)
			return err
		}
	}

	logs.Infof("%s 处理完成 %d 条数据", s.Name(), len(rows))
	return nil
}

func (s *RedisStreamEndpoint) Stock(rows []*model.RowRequest) int64 {
	var sum int64
	for _, row := range rows {
		rule, _ := global.RuleIns(row.RuleKey)
		if rule.TableColumnSize != len(row.Row) {
			logs.Warnf("%s schema mismatching", row.RuleKey)
			continue
		}

		err := s.doRuleConsume(row, rule)
		if err != nil {
			logs.Errorf(errors.ErrorStack(err))
			alarm.EmergencyAlarm(err)
			break
		}
		sum++
	}

	return sum
}

func (s *RedisStreamEndpoint) doRuleConsume(req *model.RowRequest, rule *global.Rule) error {
	kvm := rowMap(req, rule, false)

	resp := map[string]any{}
	resp["action"] = req.Action
	resp["timestamp"] = req.Timestamp
	resp["data"] = encodeValue(rule, kvm)
	resp[constants.TraceId] = stringutil.UUID()

	if rule.ReserveRawData && canal.UpdateAction == req.Action {
		old := oldRowMap(req, rule, false)
		resp["raw"] = encodeValue(rule, old)
	} else {
		resp["raw"] = ""
	}

	_, err := s.publish(rule.RedisStreamName, resp)
	if err != nil {
		logs.Errorf("Pub Message 错误 | stream: %s | error: %v", rule.RedisStreamName, err)
		return err
	}

	logs.Infof("stream name: %s, message: %+v", rule.RedisStreamName, resp)

	return err
}

func (s *RedisStreamEndpoint) publish(stream string, values map[string]any) (string, error) {
	return redisdk.XAddValues(stream, values)
}
